package com.streamsets.datacollector.pileline.stage;

import com.streamsets.datacollector.bundles.Constants;
import com.streamsets.datacollector.model.jdbc.DestinationJdbcConfigJson;
import com.streamsets.datacollector.pileline.enums.SourceConfigEnum;
import com.streamsets.datacollector.restapi.bean.ConfigConfigurationJson;
import com.streamsets.datacollector.restapi.bean.StageConfigurationJson;
import org.apache.commons.lang.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * 目标为关系型数据库(如mysql)的增量合并配置
 * @author wengzhonghui
 * @date 2019/10/16 19:12
 */
public class InitTargetOfJdbcMergeConfig {

    /*
     * 数据源stage实例名称
     */
    public static final String INSTANCE_NAME = "JDBCQuery_03";

    /*
     * SQL 查询语句节点名称
     */
    public static final String SQL_QUERY_NAME = "config.queries";

    public static void init(StageConfigurationJson stage, DestinationJdbcConfigJson destinationJdbcConfig
						, Boolean isIncMode, String updateColumn, String priColumn){
        if(INSTANCE_NAME.equalsIgnoreCase(stage.getInstanceName())) {
            if (destinationJdbcConfig != null) {
                List<ConfigConfigurationJson> configs = intConfig(stage.getConfiguration(), destinationJdbcConfig, isIncMode,updateColumn,  priColumn);
                stage.setConfiguration(configs);
            }
        }
    }

    public static List<ConfigConfigurationJson> intConfig(List<ConfigConfigurationJson> oldConfigs, DestinationJdbcConfigJson destinationJdbcConfig
						, Boolean isIncMode, String updateColumn, String priColumn){
        List<ConfigConfigurationJson> newConfigs = new ArrayList<>();
        for(ConfigConfigurationJson config :oldConfigs ){
            config = initConnectionString(config, destinationJdbcConfig);
            config = initUserName(config, destinationJdbcConfig);
            config = initPassword(config, destinationJdbcConfig);
            config = initDriverClassName(config, destinationJdbcConfig);
            config = initMergeQuery(config, destinationJdbcConfig,isIncMode, updateColumn,  priColumn);
            newConfigs.add(config);
        }
        return newConfigs;
    }


    // 生成合并sql
    private static ConfigConfigurationJson initMergeQuery(ConfigConfigurationJson config, DestinationJdbcConfigJson destinationJdbcConfig
						, Boolean isIncMode, String updateColumn, String priColumn){

        if(SQL_QUERY_NAME.equalsIgnoreCase(config.getName())){
						String targetTbName = destinationJdbcConfig.getTargetTable();
						// 增量模式
            List<String> querys = null;
						if(isIncMode!= null && isIncMode){
								querys = getQuerysForIncMode(targetTbName, updateColumn, priColumn);
						}else{
								// 全量模式
								querys = getQueryForFullMode(destinationJdbcConfig.getTargetTable());
						}
            config = new ConfigConfigurationJson(SQL_QUERY_NAME ,  querys);
        }
        return config;
    }

    /*
     * 获取增量合并的SQL
     * @param [targetTbName, updateColumn]
     * @return
     */
    private  static List<String> getQueryForFullMode(String targetTbName){
				String tableInc = "tmp_"  + targetTbName;
        List<String> queries = new ArrayList<>();
        queries.add("DROP table IF EXISTS "+ targetTbName);
        queries.add("ALTER TABLE " + tableInc + " RENAME TO  " + targetTbName);
        queries.add("CREATE TABLE " + tableInc + " LIKE " + targetTbName);

        return queries;
    }


		/**
		 * 获取增量合并的SQL
		 * @param targetTbName, updateColumn
		 * @return
		 */
		private  static List<String>  getQuerysForIncMode(String targetTbName, String updateColumn, String priColumn){
        List<String> queries = new ArrayList<>();
        // 采集的增量数据放在这个表
				String tableInc = "tmp_" + targetTbName ;
				// 增量表先把数据临时放在这里，然后再去接收新的数据
        String tableIncTemp = "tmp_"  + targetTbName + "_inc";
				// 已采集的全量表和增量表合并的内容放这里
				String mergeTableTemp = "tmp_merge_" + targetTbName;
        queries.add("create table  if not exists "+ targetTbName +" like "+ tableInc);
        queries.add("DROP TABLE IF EXISTS " + tableIncTemp);
        queries.add("ALTER TABLE "+ tableInc +" RENAME TO " + tableIncTemp);
        queries.add("CREATE TABLE  if not exists "+ tableInc +" LIKE "+ targetTbName);
        queries.add("CREATE TABLE  if not exists "+ mergeTableTemp +" LIKE "+ targetTbName);
        queries.add("TRUNCATE TABLE "+ mergeTableTemp );
        queries.add(getSqlForIncMerge(targetTbName,mergeTableTemp, tableIncTemp, updateColumn, priColumn));
        queries.add("ALTER TABLE "+ mergeTableTemp +" RENAME TO "+ targetTbName);
        queries.add("DROP TABLE IF EXISTS " + tableIncTemp );
        queries.add("DROP TABLE IF EXISTS " + mergeTableTemp );
				return queries;
		}

		/**
		 * 获取增量合并的SQL
		 * @param targetTbName, updateColumn
		 * @return
		 */
		private static String getSqlForIncMerge(String targetTbName, String mergeTableTemp, String tableIncTemp, String updateColumn, String priColumn){
				priColumn = StringUtils.isEmpty(priColumn) ? "id" : priColumn;
				//构建增量合并SQL，完整语句如：create table base_temp as SELECT t1.* FROM (SELECT * FROM base_full UNION ALL SELECT * FROM base_inc_temp) t1 JOIN(SELECT id, max(modifieddate) max_modified FROM(SELECT * FROM base_full UNION ALL SELECT * FROM base_inc_temp) t2 GROUP BY id) s ON t1.id = s.id AND t1.modifieddate = s.max_modified
				StringBuilder mergeSql = new StringBuilder();
				mergeSql.append("INSERT INTO  ");
				mergeSql.append(mergeTableTemp);
				mergeSql.append(" ( SELECT t1.* FROM (SELECT * FROM  ");
				mergeSql.append(targetTbName);
				mergeSql.append("   UNION  SELECT * FROM  ");
				mergeSql.append(tableIncTemp);
				mergeSql.append(" ) t1 JOIN(SELECT "+priColumn+", max(");
				mergeSql.append(updateColumn);
				mergeSql.append(") max_modified FROM(SELECT * FROM ");
				mergeSql.append(targetTbName);
				mergeSql.append(" UNION ALL SELECT * FROM ");
				mergeSql.append(tableIncTemp);
				mergeSql.append(") t2 GROUP BY "+priColumn+") s ON t1."+priColumn+" = s."+priColumn+" AND t1.");
				mergeSql.append(updateColumn);
				mergeSql.append(" = s.max_modified )");
				return mergeSql.toString();
		}

    // 初始化数据库连接
    private static ConfigConfigurationJson initConnectionString(ConfigConfigurationJson config, DestinationJdbcConfigJson sourceJdbcConfig){

        if(SourceConfigEnum.config_jdbcConnectionString.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(SourceConfigEnum.config_jdbcConnectionString.getName()
                    ,sourceJdbcConfig.getConnectionString());
        }
        return config;
    }

    // 初始化数据库用户名
    private static ConfigConfigurationJson initUserName(ConfigConfigurationJson config, DestinationJdbcConfigJson sourceJdbcConfig){

        if(SourceConfigEnum.config_username.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(SourceConfigEnum.config_username.getName()
                    ,sourceJdbcConfig.getJdbcUserName());
        }
        return config;
    }

    // 初始化数据库密码
    private static ConfigConfigurationJson initPassword(ConfigConfigurationJson config, DestinationJdbcConfigJson sourceJdbcConfig){

        if(SourceConfigEnum.config_password.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(SourceConfigEnum.config_password.getName()
                    ,sourceJdbcConfig.getJdbcPassword());
        }
        return config;
    }

    // 初始化数据库驱动
    private static ConfigConfigurationJson initDriverClassName(ConfigConfigurationJson config, DestinationJdbcConfigJson sourceJdbcConfig){

        if(SourceConfigEnum.driverClassName.getName().equalsIgnoreCase(config.getName())){
            config = new ConfigConfigurationJson(SourceConfigEnum.driverClassName.getName()
                    ,sourceJdbcConfig.getDriverClassName());
        }
        return config;
    }

}
